package cn.lsh.spark.streaming.kafka;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class SparkStreamingKafkaSender {

	public static void main(String[] args) throws InterruptedException {
		Properties properties = new Properties();
		properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node00:9092,node01:9092,node02:9092");
		properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerSerializer");
		properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
		properties.setProperty(ProducerConfig.ACKS_CONFIG, "-1");
		properties.setProperty(ProducerConfig.RETRIES_CONFIG, "3");
		KafkaProducer<Integer, String> producer = new KafkaProducer<>(properties);

		int i = 0;
		while (true) {
			ProducerRecord<Integer, String> record = new ProducerRecord<>("stream_2", i++, "hello_" + System.currentTimeMillis());
			producer.send(record, (metadata, e) -> {
				if (e == null) {
					System.out.println(metadata.offset() + " " + metadata.partition());
				} else {
					e.printStackTrace();
				}
			});
			if (0 == i % 2) {
				Thread.sleep(2000);
			}
		}
	}
}
